package club.janna.jrpc.cluster;

import club.janna.jrpc.cluster.loadbalance.LoadBalancer;
import club.janna.jrpc.register.Register;
import club.janna.jrpc.register.SubscribeCallback;
import club.janna.jrpc.rpc.Invoker;
import club.janna.jrpc.rpc.RPCInvocation;
import club.janna.jrpc.rpc.RPCResponse;
import club.janna.jrpc.util.SPIServiceProvider;
import lombok.extern.slf4j.Slf4j;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 集群调用者
 * @author Panbo.Guo
 */
@Slf4j
public class ClusterInvoker implements Invoker {

    /**
     * 注册器
     */
    private final Register register;
    /**
     * 服务的名字
     */
    private final String serviceName;
    /**
     * 调用者Map
     */
    private Map<URI, Invoker> invokerMap = new ConcurrentHashMap<>();

    /**
     * 负载均衡器
     */
    private LoadBalancer loadBalancer;

    /**
     *
     * @param registry 注册中心地址
     * @param serviceName 服务名称
     * @param loadBalancer 负载均衡策略
     */
    public ClusterInvoker(String registry, String serviceName, String loadBalancer) {
        this.serviceName = serviceName;
        URI registryURI;
        try {
            registryURI = new URI(registry);
        } catch (URISyntaxException e) {
            throw new RuntimeException("注册中心配置错误");
        }
        // 初始化注册中心
        this.register = SPIServiceProvider.getRegister(registryURI);

        // 获取invoker并添加到Map中
        Set<URI> providers = this.register.fetch(serviceName);
        providers.forEach(uri -> this.invokerMap.put(uri, getInvoker(uri)));

        // 初始化负载均衡器
        this.loadBalancer = SPIServiceProvider.getLoadBalancer(loadBalancer);
        this.initLoadBalance();

        try {
            // 注册消费者
            this.register.registerConsumer(serviceName, new URI(String.format("http://xxx:8080/%s?loadBalance=%s", serviceName, loadBalancer)));
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }

        // 订阅服务更新
        this.register.subscribe(serviceName, (uri, action) -> {
            if(action == SubscribeCallback.Action.UPDATE) {
                if(invokerMap.containsKey(uri))
                    return;
                log.info("发现新的服务提供者[{}], uri -> {}", serviceName, uri.toString());
                invokerMap.put(uri, getInvoker(uri));
            } else if(action == SubscribeCallback.Action.REMOVE) {
                if(!invokerMap.containsKey(uri))
                    return;
                log.info("服务提供者已下线[{}], uri -> {}", serviceName, uri.toString());
                Invoker invoker = invokerMap.remove(uri);

                // 如果需要关闭，则关闭该invoker
                if(invoker instanceof AutoCloseable) {
                    try {
                        ((AutoCloseable) invoker).close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
            initLoadBalance();
        });
    }


    @Override
    public RPCResponse<?> invoker(RPCInvocation invocation) {
        try {
            return loadBalancer.next().invoker(invocation);
        } catch (Exception e) {
            log.error("调用服务失败，invocation -> {}", invocation, e);
            throw e;
        }
    }

    /**
     * 通过URI获取Invoker
     * @param uri
     * @return
     */
    private Invoker getInvoker(URI uri) {
        return SPIServiceProvider.getProtocol(uri.getScheme()).buildInvoker(uri);
    }

    /**
     * 初始化负载均衡器
     */
    private void initLoadBalance() {
        Invoker[] invokers = new Invoker[this.invokerMap.size()];
        this.loadBalancer.init(this.invokerMap.values().toArray(invokers));
    }
}
